/*
   Copyright 2009 Ryan Phillips

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
#include <apr.h>
#include <apr_network_io.h>
#include <apr_poll.h>
#include <apr_buckets.h>
#include <apr_strings.h>
#include <apr_getopt.h> 
#include <iostream>
#include <sstream>
#include <error.h>
#include <sys/inotify.h>
#include "CoreMessage.pb.h"
#include "remotecores.h"

using namespace std;

/* size of the event structure, not counting name */
#define EVENT_SIZE (sizeof (struct inotify_event))

/* reasonable guess as to size of 1024 events */
#define BUF_LEN (1024 * (EVENT_SIZE + 16))

void watch_for_cores(apr_pool_t *pool, const char *path, const char *hostname,
                     apr_port_t port);

static void usage(apr_pool_t *pool);

static const apr_getopt_option_t client_options[] = {
    {"remote",    'r', 1, "FQDN or IP of remote server"},
    {"port",      'p', 1, "Port of server"},
    {"dir",       'd', 1, "Directory to watch"},
    {0, 0, 0, 0},
};


int main(int argc, const char **argv)
{
    apr_pool_t *g_pool;
    apr_getopt_t *os;
    apr_status_t rv;
    int opt;
    const char *arg;
    const char *path = NULL;
    const char *host = NULL;
    apr_port_t port = REMOTE_CORES_DEFAULT_PORT;

    /* Init */
    apr_initialize();
    apr_pool_create(&g_pool, NULL);

    /* Parse Commandline */
    rv = apr_getopt_init(&os, g_pool, argc, argv);
    if (rv != APR_SUCCESS) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        error(rv, 0, "apr_getopt_init : %s", buffer);
    }

    while (1) {
        rv = apr_getopt_long(os, client_options, &opt, &arg);
        if (APR_STATUS_IS_EOF(rv)) {
            break;
        }

        if (rv != APR_SUCCESS) {
            usage(g_pool);
        }

        switch (opt) {
        case 'r':
            host = arg;
            break;
        case 'p':
            port = atoi(arg);
            break;
        case 'd':
            path = arg;
            break;
        }
    }

    if (path == NULL || host == NULL) {
        usage(g_pool);
    }

    fprintf(stderr, "remotecore-client: Using Server (%s:%i)\n", host, port);
    fprintf(stderr, "remotecore-client: Watching Path (%s)\n", path);

    /* Run */
    watch_for_cores(g_pool, path, host, port);

    /* Cleanup */
    apr_pool_destroy(g_pool);
    apr_terminate();

    return 0;
}

/* Return the fd for the inotify device */
static int init_watch(const char *path)
{
    int fd;
    int rv;

    fd = inotify_init();
    if (fd < 0) {
        perror("inotify_init");
    }

    rv = inotify_add_watch(fd, path, IN_CLOSE_WRITE);
    if (rv < 0) {
        perror("inotify_add_watch");
    }
    
    return fd;
}

static void send_brigade(apr_socket_t *sock, apr_bucket_brigade *bb)
{
    apr_status_t rv;
    apr_bucket* bucket = APR_BRIGADE_FIRST(bb);
    while (bucket != APR_BRIGADE_SENTINEL(bb)) {
        const char* buffer;
        apr_size_t nbytes, o, i;

        apr_bucket_read(bucket, &buffer, &nbytes, APR_BLOCK_READ);

        o = 0;
        i = nbytes;
        while (i > 0) {
            nbytes = i;

            rv = apr_socket_send(sock, buffer + o, &nbytes);
            if (rv != APR_SUCCESS) {
                char buffer[256];
                apr_strerror(rv, buffer, sizeof(buffer));
                error(rv, 0, "%s", buffer);
            }

            o +=  nbytes;
            i -= nbytes;
        } 
        
        bucket = APR_BUCKET_NEXT (bucket);
    }
}

static void send_core(apr_pool_t *pool, const char *path,
                      const char *filename, const char *hostname, 
                      apr_port_t port)
{
    apr_pool_t *t_pool;
    apr_socket_t *sock;
    apr_sockaddr_t* address;
    apr_finfo_t finfo;
    apr_file_t *file;
    apr_bucket_alloc_t *alloc;
    apr_bucket_brigade *out;
    apr_size_t len;
    char *filepath;
    apr_status_t rv;
    apr_int32_t off = 0;
    CoreMessage msg;

    apr_pool_create(&t_pool, pool);

    filepath = apr_psprintf(t_pool, "%s/%s", path, filename);

    rv = apr_socket_create(&sock, APR_INET, SOCK_STREAM, APR_PROTO_TCP, t_pool);
    if (rv != APR_SUCCESS) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        error(rv, 0, "%s", buffer);
    }

    /* Set blocking */
    apr_socket_opt_set(sock, APR_SO_NONBLOCK, off);

    /* Connect to IP */
    rv = apr_sockaddr_info_get(&address, hostname, APR_UNSPEC, port, 0, t_pool);
    if (rv != APR_SUCCESS) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        return;
    }

    rv = apr_socket_connect(sock, address);

    if (rv != APR_SUCCESS &&
        !APR_STATUS_IS_EAGAIN(rv) &&
        !APR_STATUS_IS_EINPROGRESS(rv)) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        return;
    }

    alloc = apr_bucket_alloc_create(t_pool);

    rv = apr_file_open(&file, filepath, APR_READ, APR_FPROT_OS_DEFAULT, 
                       t_pool);
    if (rv != APR_SUCCESS) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        error(rv, 0, "%s", buffer);
    }

    apr_file_info_get(&finfo, APR_FINFO_SIZE, file);

    /* Output Header and marshall protobuffer */
    msg.set_filename(filename);
    msg.set_filesize(finfo.size);

    stringstream output(stringstream::in | stringstream::out);
    msg.SerializeToOstream(&output);

    struct iovec vec[2];
    apr_uint32_t pb_length = htonl(output.str().length());

    vec[0].iov_base = &pb_length;
    vec[0].iov_len = sizeof(pb_length);
    vec[1].iov_base = (void*) output.str().c_str();
    vec[1].iov_len = output.str().length();
    rv = apr_socket_sendv(sock, vec, 2, &len);
    if (rv != APR_SUCCESS) {
        char buffer[256];
        apr_strerror(rv, buffer, sizeof(buffer));
        error(rv, 0, "%s", buffer);
    }

    /* Send Core */
    out = apr_brigade_create(t_pool, alloc);
    APR_BRIGADE_INSERT_TAIL(out,
                            apr_bucket_file_create(file,
                                                   0,
                                                   finfo.size,
                                                   t_pool,
                                                   alloc));
    send_brigade(sock, out);

    apr_bucket_alloc_destroy(alloc);

    apr_pool_destroy(t_pool);
}

void watch_for_cores(apr_pool_t *pool, const char *path, const char *hostname, apr_port_t port)
{
    int inotify_fd;
    int len;
    int i;
    char buf[BUF_LEN];

    inotify_fd = init_watch(path);

    while (1) {
        len = read(inotify_fd, buf, BUF_LEN);
        if (len < 0) {
            if (errno == EINTR) {
                continue;
            }
            else {
                perror("read");
            }
        }

        i = 0;
        while (i < len) {
            struct inotify_event *event;

            event = (struct inotify_event *) &buf[i];
            if (event->len && strstr(event->name, "core")) {
                /* send the core */
                fprintf(stderr, "remotecore-client: Found Core (%s)\n", event->name);
                send_core(pool, path, event->name, hostname, port);
            }

            i += EVENT_SIZE + event->len;
        }
    }

    close(inotify_fd);
}

static void usage(apr_pool_t *pool)
{
    (void) pool;
    fprintf(stderr, "usage: remotecore-client -r SERVER_IP [-p PORT] --dir /path/to/directory\n");
    exit(EXIT_FAILURE);
}

